package bch.rocketmq.controller;

import bch.rocketmq.A;
import bch.rocketmq.B;
import bch.rocketmq.producer.MQProducer;
import bch.rocketmq.service.BoundedContainer;
import bch.rocketmq.util.SnowflakeWorker;
import bch.rocketmq.service.CompanyScannerService;
import bch.rocketmq.service.OrderDelayedQueueService;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.redisson.api.*;
import org.redisson.spring.transaction.RedissonTransactionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.jdbc.support.JdbcTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;

import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/rocketmq")
@RequiredArgsConstructor
public class RocketMQController {

    private static final Logger log = LoggerFactory.getLogger(RocketMQController.class);
    private final MQProducer mqProducer;
    private final ApplicationContext applicationContext;
    private final JdbcClient jdbcClient;
    private final RedissonClient redissonClient;

    @GetMapping("/sendTag")
    public SendResult sendTag() {
        return mqProducer.sendTagMsg("带有tag的字符消息");
    }

    @GetMapping("/sendAsync")
    public void sendAsync() {
        mqProducer.sendAsyncMsg("user");
    }

    @GetMapping("/sendOneWay")
    public void sendOneWay() {
        mqProducer.sendOneWayMsg("user");
    }

    @GetMapping("/sendDelay")
    public void sendDelay() {
        mqProducer.sendDelayMsg("延迟消息", 2);
    }

    @GetMapping("/sendOrderly")
    public void sendOrderly() {
        for (int i = 0; i < 100; i++) {
            mqProducer.sendOrderMsg("顺序消息" + i);
        }
    }
    @GetMapping("/sendTransactional")
    public void sendTransactional(String id) {
        TransactionSendResult transactionSendResult = mqProducer.sendTransactionMsg("事务消息" + id);
        LocalTransactionState localTransactionState = transactionSendResult.getLocalTransactionState();
    }

    @Autowired
    private RedissonTransactionManager redissonTransactionManager;
    @Autowired
    private JdbcTransactionManager jdbcTransactionManager;

    /**
     * a. 部分提交风险
     *
     * ChainedTM 并不是 XA 事务，提交是顺序的。
     *
     * 如果 Redisson 提交成功后，JDBC 提交失败：
     *
     * Redisson 已经 commit，JDBC rollback → 数据不一致
     *
     * Spring 会抛 HeuristicCompletionException 提示“部分提交”。
     *
     * 所以“都能 rollback”只在方法内部抛异常或提交前失败时才成立。
     */
    @GetMapping("/sendTransaction4Redis")
    // 默认使用的redissionTransactionManager
    @Transactional(rollbackFor = Exception.class, transactionManager = "chainedTransactionManager")
    public void sendTransaction4Redission(String id) {
        jdbcClient.sql("insert into company values(?,?,?)").params(SnowflakeWorker.newId(), "name", SnowflakeWorker.newId()).update();

//        int i = 1 / 0;

        RTransaction transaction = redissonTransactionManager.getCurrentTransaction();
        RMap<String, String> map = transaction.getMap("test1");
        map.put("1", "value");


        // 自动commit rollback
//        int i = 1 / 0;
    }

    private final DeferredResult<String> deferredResult = new DeferredResult<>();
    @GetMapping("/slow-api")
    public DeferredResult<String> getSlowResponse() {
        System.out.println("主线程接收请求，开始异步处理...");

        DeferredResult<String> deferredResult = new DeferredResult<>(5000L, "请求处理超时。");

        // 注册超时处理回调
        deferredResult.onTimeout(() -> {
            System.out.println("DeferredResult 超时处理回调被调用。");
            // 在这里可以设置一个超时结果，或者执行其他清理逻辑
            // deferredResult.setErrorResult("请求超时，请稍后重试。");
        });

        new Thread(() -> {
            try {
                Thread.sleep(10000);
                System.out.println("异步任务完成，尝试设置结果...");
                deferredResult.setResult("任务成功完成！");
            } catch (InterruptedException e) {
                System.out.println("异步任务被中断！");
            }
        }).start();

        System.out.println("主线程已返回，等待异步结果...");
        return deferredResult;
    }

    @Autowired
    private OrderDelayedQueueService delayQueueService;

    @PostMapping("/add-order")
    public String addOrder(@RequestParam(defaultValue = "10") long delay) {
        String orderId = "order-" + UUID.randomUUID().toString().substring(0, 8);
        delayQueueService.addOrder(orderId, delay, TimeUnit.SECONDS);
        return "订单 " + orderId + " 已添加到延迟队列，将在 " + delay + " 秒后处理。";
    }

    @PostMapping("tryLock")
    public String tryLock() {
        RLock lock = redissonClient.getLock("order_lock");
        if (lock.tryLock()) {
            try {
                // 模拟处理逻辑
                Thread.sleep(30000);
            } catch (InterruptedException e) {
                // GracefulShutdown等待一段时间（默认 30s）后，会去中断还在执行的请求线程
                log.error("处理订单失败");
                Thread.currentThread().interrupt();
            } finally {
                lock.unlock();
            }
        }else {
            return "处理失败，锁已被占用";
        }
        return "处理完成";
    }

    @Autowired
    private CompanyScannerService companyScannerService;
    @PostMapping("scanCompany")
    public String scanCompany() {
        companyScannerService.scanCompaniesMultiThread(2);
        return "扫描完成";
    }

    @Autowired
    private RRateLimiter rateLimiter;
    @GetMapping("sentinelByRedis")
    public String sentinelByRedis() {
        // 获取令牌
        boolean allowed = rateLimiter.tryAcquire();
        if (allowed) {
            return "成功执行业务";
        } else {
            return "限流了，请稍后再试";
        }
    }

    @Autowired
    private BoundedContainer boundedContainer;

    @PostMapping("boundedContainer")
    public String boundedContainer(@RequestParam String ele) {
        try {
            boundedContainer.put(ele);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "执行线程被打断";
        }
        return "执行完成";
    }



    @GetMapping("boundedContainer")
    public Object boundedContainer() {
        System.out.println(Thread.currentThread().getContextClassLoader());
        try {

            return boundedContainer.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "执行线程被打断";
        }
    }
}
